package adapter

import (
	"bytes"
	"context"
	"encoding/json"
	amqp "github.com/rabbitmq/amqp091-go"
	"go.uber.org/zap"
	"time"
)

const (
	ContentTypeJson = "application/json"
	ContentEncoding = "UTF-8"
	AppId           = "go-amqp-client"
)

type Client struct {
	Conn   *AmqpConn
	Logger *zap.SugaredLogger
}

func (cli *Client) SetLogger(log *zap.SugaredLogger) {
	cli.Logger = log
}

func NewClient(conn *AmqpConn) *Client {
	return &Client{Conn: conn}
}

// initQueueAndMessage 初始化队列和消息
func (cli *Client) initQueueAndMessage(message *Message) error {
	exchange := "" // 默认交换机
	queueName := message.QueueName

	// 创建普通队列
	if err := cli.createQueue(queueName); err != nil {
		cli.Logger.Errorf("falied to declare queue: [%s], err: %s\n", queueName, err)
		return err
	}

	if message.DelayTime > 0 {
		// 延时消息模式, 需要rabbitmq安装插件: delayed_message_exchange
		// 声明交换机
		exchange = "delayed_exchange"
		if err := cli.createDelayExchange(exchange); err != nil {
			cli.Logger.Errorf("falied to declare delayed exchange: [%s], err: %s\n", exchange, err)
			return err
		}

		// 队列绑定到延时交换机
		if err := cli.queueBind(queueName, queueName, exchange); err != nil {
			cli.Logger.Errorf("falied to bind queue to delayed exchange: [%s -> %s], err: %s\n", queueName, exchange, err)
			return err
		}

		// 设置消息延时属性
		message.MsgHeaders["x-delay"] = message.DelayTime / time.Millisecond // 单位:ms
	}

	// 设置消息属性
	message.Exchange = exchange
	message.RoutingKey = queueName
	return nil
}

func (cli *Client) createQueue(queueName string) error {
	_, err := cli.Conn.channel.QueueDeclare(queueName, true, false, false, false, nil)
	return err
}

func (cli *Client) queueBind(queueName, routingKey, exchange string) error {
	return cli.Conn.channel.QueueBind(queueName, routingKey, exchange, false, nil)
}

func (cli *Client) createDelayExchange(exchange string) error {
	// 需要rabbitmq安装插件: delayed_message_exchange
	delayedExchangeType := "x-delayed-message" // 延时消息交换机类型
	amqpTable := amqp.Table{
		"x-delayed-type": amqp.ExchangeDirect, // 实际的交换机的类型在这里指定
	}
	return cli.Conn.channel.ExchangeDeclare(exchange, delayedExchangeType, true, false, false, false, amqpTable)
}

// 发布消息
func (cli *Client) publishMessage(message *Message) error {
	// 组装消息参数
	params := make(map[string]interface{})
	params["route"] = message.Route
	params["param"] = message.Params

	// json_encode
	bts := bytes.NewBuffer([]byte{})
	err := json.NewEncoder(bts).Encode(params)
	if err != nil {
		cli.Logger.Errorf("falied to encode msg, err: %s\n", err)
		return err
	}

	// 设置消息属性
	msg := amqp.Publishing{
		Headers:         message.MsgHeaders,
		ContentType:     ContentTypeJson,
		ContentEncoding: ContentEncoding,
		MessageId:       message.MsgId,
		Timestamp:       time.Now(),
		AppId:           AppId,
		Body:            bts.Bytes(),
	}
	return cli.Conn.channel.PublishWithContext(context.Background(), message.Exchange, message.RoutingKey, false, false, msg)
}
